package com.zhang.gmall.utils;

import com.zhang.gmall.beans.TransientSink;
import com.zhang.gmall.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @title: clickhouse工具类
 * @author: zhang
 * @date: 2022/3/27 08:06
 */
public class ClickHouseUtil {

    public static <T> SinkFunction<T> getClickHouseSink(String sql) {
        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement statement, T t) throws SQLException {
                        //通过反射方式获取泛型对象中的所有属性
                        Class<?> clz = t.getClass();
                        Field[] fields = clz.getDeclaredFields();
                        //通过遍历所有的属性，获取对应的属性值，并赋给预编译sql的占位符
                        int offset = 0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];
                            //设置私有属性可以访问
                            field.setAccessible(true);
                            try {
                                Object fieldValue = field.get(t);
                                //遇到不需要赋值的字段跳过
                                if (field.getAnnotation(TransientSink.class) != null) {
                                    offset++;
                                    continue;
                                }
                                statement.setObject(i + 1 - offset, fieldValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                                System.out.println("获取属性之失败～～～～～");
                            }

                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(5000L)
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
    }
}
